Workflowsが多重起動されないようにする方法をCloud Run Functionsを用いる方法で考えてみた

import functions_framework
from google.cloud.workflows.executions_v1 import ExecutionsClient
from google.cloud.workflows.executions_v1.types import ListExecutionsRequest, Execution
from flask import jsonify, abort

def can_execute_workflow(project_num: str, location: str, workflow_name: str, current_execution_name: str):
    """
    指定されたワークフローの多重起動判定関数
    現在の実行がACTIVE状態の中で最も開始時間が古いかどうかを判定

    Args:
        project_num (str): プロジェクト番号
        location (str): ワークフローがデプロイされているリージョン
        workflow_name (str): ワークフローの名前
        current_execution_name (str): 現在の実行名

    Returns:
        dict: 実行可否と関連情報を含むレスポンス
            - "can_execute" (bool): 実行可能であればTrue
            - "message" (str): 実行可否の理由
            - "latest_active_execution" (dict or None): 現在のACTIVEな実行の中で最も古い実行の情報
    """
    client = ExecutionsClient()
    parent = f"projects/{project_num}/locations/{location}/workflows/{workflow_name}"

    try:
        # 実行リストを取得
        request = ListExecutionsRequest(parent=parent)
        response = client.list_executions(request=request)

        # ACTIVE状態の実行をフィルタリング
        active_executions = [
            execution for execution in response.executions
            if execution.state == Execution.State.ACTIVE  # Enumの値と比較
        ]

        # ACTIVE状態の実行がない場合実行可能
        if not active_executions:
            return {
                "can_execute": True,
                "message": "No ACTIVE executions found. Safe to execute.",
                "latest_active_execution": None,
            }

        # 最も開始時間が古いACTIVEな実行を取得
        oldest_execution = min(
            active_executions,
            key=lambda execution: execution.start_time
        )

        # 現在の実行が最も古いかどうか判定
        if oldest_execution.name == current_execution_name:
            return {
                "can_execute": True,
                "message": "Current execution is the oldest ACTIVE execution. Safe to execute.",
                "latest_active_execution": {
                    "name": oldest_execution.name,
                    "state": Execution.State(oldest_execution.state).name,
                    "start_time": oldest_execution.start_time,
                },
            }
        else:
            return {
                "can_execute": False,
                "message": "Current execution is not the oldest ACTIVE execution. Do not execute.",
                "latest_active_execution": {
                    "name": oldest_execution.name,
                    "state": Execution.State(oldest_execution.state).name,
                    "start_time": oldest_execution.start_time,
                },
            }

    except Exception as e:
        return {
            "can_execute": False,
            "message": f"Error occurred while checking executions: {e}",
            "latest_active_execution": None,
        }

# Cloud Functions HTTPエントリポイント
@functions_framework.http
def check_workflows_run(request):
    """
    Cloud FunctionsのHTTPエントリポイント
    JSONリクエストを受け取り、ワークフローの実行可否を判定

    Args:
        request (flask.Request): HTTPリクエストオブジェクト

    Returns:
        flask.Response: 実行可否の結果を含むJSONレスポンス
    """
    try:
        # JSONデータを取得
        data = request.get_json()
        if not data:
            abort(400, description="Invalid JSON payload")

        # 必須フィールドのチェック
        required_fields = ["project_num", "location", "workflow_name", "execution_id"]
        for field in required_fields:
            if field not in data:
                abort(400, description=f"Missing required field: {field}")

        # 必要な情報を取得
        project_num = data['project_num']
        location = data['location']
        workflow_name = data['workflow_name']
        execution_id = data['execution_id']
        current_execution_name = f"projects/{project_num}/locations/{location}/workflows/{workflow_name}/executions/{execution_id}"

        # 実行可否を判定
        result = can_execute_workflow(project_num, location, workflow_name, current_execution_name)

        # 実行可能の場合は200を返す
        if result["can_execute"]:
            return jsonify(result), 200

        # 実行不可の場合は503を返す
        return jsonify(result), 503

    except Exception as e:
        return jsonify({"error": f"Unexpected error: {e}"}), 500

functions-framework==3.*
google-cloud-workflows

- init:
    assign:
        - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
        - project_num: ${sys.get_env("GOOGLE_CLOUD_PROJECT_NUMBER")}
        - workflow_exec_id: ${sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
        - workflow_name: ${sys.get_env("GOOGLE_CLOUD_WORKFLOW_ID")}
        - workflow_location: ${sys.get_env("GOOGLE_CLOUD_LOCATION")}
        - functions_url: "作成したCloud Run FunctionsのURL"
- checkRunsStep:
    call: http.post
    args:
        url: ${functions_url}
        auth:
            type: OIDC
        body:
            project_num: ${project_num}
            location: ${workflow_location}
            workflow_name: ${workflow_name}
            execution_id: ${workflow_exec_id}
        timeout: 500
    result: processResult
- sleepStep:
    call: sys.sleep
    args:
        seconds: 30
- endStep:
    return: ${processResult}

#!/bin/bash

# ワークフロー名とリージョンを設定
WORKFLOW_NAME="ワークフロー名"
LOCATION="asia-northeast1"

# 5回並列実行
for i in {1..5}; do
  echo "Executing workflow $i..."
  gcloud workflows execute "$WORKFLOW_NAME" --location="$LOCATION" &
done
# 全てのバックグラウンドプロセスが完了するのを待つ
wait
echo "All executions completed."

sh xxx.sh

gcloud workflows executions list "ワークフロー名" --location=asia-northeast1